package com.itstyle.quartz.kettleUtil;

import com.alibaba.fastjson.JSONObject;
import com.itstyle.quartz.service.dao.IJobService;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.Result;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.KettleLogStore;
import org.pentaho.di.core.logging.KettleLoggingEvent;
import org.pentaho.di.core.logging.KettleLoggingEventListener;
import org.pentaho.di.core.util.EnvUtil;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Set;

public class kettleUtil {

    @Autowired
    private IJobService jobService;

    private static final Logger logger = LoggerFactory.getLogger(kettleUtil.class);
    /**
     * 本测试类慎用！！！！！！！
     *
     * @param args
     */
    public static void main(String[] args) throws Exception {
        String datetime = "2014-12-19 23:20:45";
        //String[] params = {"707", datetime}; // 传递参数
        String[] params = null;
        String path = "D:\\p1000\\多线程测试\\多线程测试.ktr";
        //runKtr(params,path);
       /* for (int i = 0; i < 3; i++) {
            params[0] += i;
            runJob(params, path);
        }*/
    }

    public static Result runKtr(String[] params, String path) throws Exception {

        Trans trans = null;

        KettleEnvironment.init();// 初始化
        EnvUtil.environmentInit();
        //增加监听，获得脚本运行时得日志
        KettleLogStore.getAppender().clear();
        KettleLogStore.getAppender().addLoggingEventListener(new KettleLoggingEventListener() {
            @Override
            public void eventAdded(KettleLoggingEvent logs) {
                //logs为kettle运行时输出的日志。
                logger.info(logs.getMessage().toString());
            }
        });

        TransMeta transMeta = new TransMeta(path);
            // 转换
            trans = new Trans(transMeta);
            trans.execute(params);
            // 等待转换执行结束
            trans.waitUntilFinished();



        // 抛出异常
        if (trans.getErrors() > 0) {
            throw new KettleException("There are errors during transformation exception!(执行ktr发生异常)");
        }

        return trans.getResult();
    }

    public static Result runJob(JSONObject params, String path)  throws Exception {

        KettleEnvironment.init();
        //增加监听，获得脚本运行时得日志
        KettleLogStore.getAppender().clear();
        KettleLogStore.getAppender().addLoggingEventListener(new KettleLoggingEventListener() {
            @Override
            public void eventAdded(KettleLoggingEvent logs) {
                //logs为kettle运行时输出的日志。
                logger.info(logs.getMessage().toString());
            }
        });
        JobMeta jobMeta = new JobMeta(path, null);
        Job job = new Job(null, jobMeta);
        if(params != null) {
            Set<String> keys = params.keySet();
            for (String key : keys) {
                System.out.println("key值：" + key + " value值：" + params.get(key));
                // 向Job 脚本传递参数，脚本中获取参数值：${参数名}
                job.setVariable(key.toString(), params.get(key).toString());
            }
        }
        job.start();

        job.waitUntilFinished();

        if (job.getErrors() > 0) {
            throw new KettleException("There are errors during job exception!(执行job发生异常)");
        }

        return job.getResult();
    }
}
